package com.intct.flink.window;

import com.intct.hbase.bean.WindowBean;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author gufg
 * @since 2025-06-27 14:26
 */
public class WindowAggreagteDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);

        // 从socket获取数据
        DataStreamSource<String> sourceDS = env.socketTextStream("jd-node", 8888);

        // Map转换(将字符转换为实体Bean)
        SingleOutputStreamOperator<WindowBean> mapDS = sourceDS.map(sources -> {
            String[] sourceList = sources.split(",");
            return WindowBean.builder()
                    .name(sourceList[0])
                    .ts(Long.valueOf(sourceList[1]))
                    .vc(Integer.valueOf(sourceList[2]))
                    .build();
        });

        // 按照指定Key分区
        KeyedStream<WindowBean, String> keyByDS = mapDS.keyBy(key -> key.getName());

        // 窗口-- 分配器(滚动窗口) 10秒一个窗口
        WindowedStream<WindowBean, String, TimeWindow> windowDS = keyByDS.window(TumblingProcessingTimeWindows.of(Time.seconds(30)));
        /**
         * [0~10)
         * [10~20)
         */

        // 窗口 -- 窗口函数
        /**
         *
         * 参数:
         *      第一参数<IN> : 聚合的值的类型(输入值)
         *      第二参数<ACC>:累加器的类型(中间聚合状态)。
         *      第三参数<OUT>:聚合结果的类型
         */
        SingleOutputStreamOperator<String> aggregateDS = windowDS.aggregate(new AggregateFunction<WindowBean, Integer, String>() {
            /**
             * 创建累加器
             * @return
             */
            @Override
            public Integer createAccumulator() {
                long currentTime = System.currentTimeMillis();
                String currentDate = DateFormatUtils.format(currentTime, "yyyy-MM-dd HH:mm:ss");
                System.out.println("createAccumulator方法 , 时间:" + currentDate);
                return 0;
            }

            /**
             * 累加
             * @param value The value to add
             * @param accumulator The accumulator to add the value to
             * @return
             */
            @Override
            public Integer add(WindowBean value, Integer accumulator) {
                System.out.println("add方法");
                return value.getVc() + accumulator;
            }

            /**
             * 返回结果
             * @param accumulator The accumulator of the aggregation
             * @return
             */
            @Override
            public String getResult(Integer accumulator) {
                System.out.println("getRueslut方法");
                return String.valueOf(accumulator);
            }

            /**
             *
             * @param a An accumulator to merge
             * @param b Another accumulator to merge
             * @return
             */
            @Override
            public Integer merge(Integer a, Integer b) {
                System.out.println("merge方法");
                return null;
            }
        });

        aggregateDS.print();

        env.execute();
    }
}
